package jmind.core.mq.kafka;

import java.util.Properties;

import jmind.base.lang.SourceProperties;
import jmind.base.util.CollectionUtil;
import jmind.base.util.GlobalConstants;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractKafkaConsumer {

    private boolean started = true;
    // 主题
    private String topicName;
    // 分组，组相同则是消息模式，只会消费一次。 组不同就是订阅模式，消费多次 
    private String groupName;

    protected KafkaConsumer<String, String> consumer;

    private final Logger logger = LoggerFactory.getLogger(getClass());

    public AbstractKafkaConsumer(String topicName, String groupName) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                SourceProperties.getDataSource().getProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG));
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupName);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
        this.topicName = topicName;
        this.groupName = groupName;
        listen();
    }

    private void listen() {

        this.consumer.subscribe(CollectionUtil.asList(topicName, GlobalConstants.COMMA));

        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                while (isStarted()) {
                    ConsumerRecords<String, String> records = consumer.poll(10000);

                    try {
                        handleMessage(records);
                        consumer.commitAsync();
                    } catch (Exception e) {
                        logger.error("consumer error, topic={},groupName={}", topicName, groupName, e);
                    }
                }
            }
        };

        Thread thread = new Thread(runnable);
        thread.setDaemon(true);
        thread.setName(this.groupName + "-" + groupName + thread.getName());
        thread.start();
        logger.info("topic=" + topicName + ", groupName=" + groupName + " started");
    }

    public abstract void handleMessage(ConsumerRecords<String, String> records) throws Exception;

    public String getTopicName() {
        return topicName;
    }

    public void setTopicName(String topicName) {
        this.topicName = topicName;
    }

    public String getGroupName() {
        return groupName;
    }

    public void setGroupName(String groupName) {
        this.groupName = groupName;
    }

    public boolean isStarted() {
        return started;
    }

    public void setStarted(boolean started) {
        this.started = started;
    }

}
